package spring.cloud.tasks.api.service.impl;

import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import spring.cloud.tasks.api.database.repository.NameSpaceDtoRepository;
import spring.cloud.tasks.api.database.repository.NameSpaceZooKeeperClusterMappingDtoRepository;
import spring.cloud.tasks.api.database.repository.TaskConfigurationDtoRepository;
import spring.cloud.tasks.api.domain.ExecutorInfo;
import spring.cloud.tasks.api.domain.NameSpaceConfiguration;
import spring.cloud.tasks.api.domain.TaskConfiguration;
import spring.cloud.tasks.api.domain.TaskExecutorStatus12;
import spring.cloud.tasks.api.service.ExecutorService;
import spring.cloud.tasks.api.service.NameSpaceService;
import spring.cloud.tasks.api.service.TaskService;
import spring.cloud.tasks.api.service.ZooKeeperService;
import spring.cloud.tasks.api.zookeeper.ZooKeeperRepository;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class NameSpaceServiceImpl implements NameSpaceService {

    private static final Logger log = LoggerFactory.getLogger(NameSpaceServiceImpl.class);

    @Autowired
    private TaskService taskService;

    @Autowired
    private ExecutorService executorService;

    @Autowired
    private NameSpaceDtoRepository namespaceInfoRepository;

    @Autowired
    private TaskConfigurationDtoRepository currentJobConfigRepository;

    @Autowired
    private NameSpaceZooKeeperClusterMappingDtoRepository namespaceZkClusterMappingRepository;

    @Autowired
    private ZooKeeperService zooKeeperService;

    @Autowired
    private ZooKeeperRepository curatorRepository;

    //@Override
    public Map<String, List> importTasksFromNameSpaceToNameSpace(String srcNamespace, String destNamespace,
                                                                 String createdBy) throws RuntimeException {

        if (StringUtils.isBlank(srcNamespace)) {
            throw new RuntimeException("srcNamespace should not be null");
        }
        if (StringUtils.isBlank(destNamespace)) {
            throw new RuntimeException("destNamespace should not be null");
        }
        if (StringUtils.equals(srcNamespace, destNamespace)) {
            throw new RuntimeException(
                    "srcNamespace and destNamespace should be difference");
        }

        try {
            List<String> successfullyImportedJobs = new ArrayList<>();
            List<String> failedJobs = new ArrayList<>();
            Map<String, List> result = new HashMap<>();
            result.put("success", successfullyImportedJobs);
            result.put("fail", failedJobs);

            List<TaskConfiguration> taskConfigurationList = taskService.getTaskConfigurationList(srcNamespace);

            for (TaskConfiguration taskConfiguration : taskConfigurationList) {
                String taskName = taskConfiguration.getTaskId();
                try {
                    taskService.addJob(destNamespace, taskConfiguration, createdBy);
                    successfullyImportedJobs.add(taskName);
                } catch (RuntimeException e) {
                    log.warn("fail to import job {} from {} to {}", taskName, srcNamespace, destNamespace, e);
                    failedJobs.add(taskName);
                }
            }

            return result;
        } catch (RuntimeException e) {
            log.warn("import jobs from {} to {} fail", srcNamespace, destNamespace, e);
            throw e;
        }
    }

    //@Override
    public void deleteNamespace(String namespace) throws RuntimeException {
        boolean online = isExecutorsOnline(namespace);
        if (online) {
            log.info("namespace {} has online executor, can not delete it", namespace);
            throw new RuntimeException(
                    "namespace has online executor, can not delete it");
        } else {
            NameSpaceConfiguration namespaceConfiguration = zooKeeperService.getNameSpaceConfigurationByNameSpace(namespace);
            deleteInfosInDB(namespace);
            deleteNameSpaceInZooKeeper(namespaceConfiguration, namespace);
        }
    }


    protected boolean isExecutorsOnline(String namespace) throws RuntimeException {
        List<ExecutorInfo> executors = executorService.getExecutorInfoList(namespace, TaskExecutorStatus12.ONLINE);
        return executors.size() > 0;
    }

    @Transactional(rollbackFor = Exception.class)
    protected void deleteInfosInDB(String namespace) {
        deleteNameSpaceInDB(namespace);
        log.info("delete namespace in DB success - namespace {}", namespace);
        deleteNameSpaceZooKeeperClusterMappingInDb(namespace);
        log.info("delete namespaceZkCluster in DB success - namespace {}", namespace);
        deleteJobsInDB(namespace);
        log.info("delete jobs in DB success - namespace {}", namespace);
    }

    protected void deleteNameSpaceInDB(String nameSpace) {
        namespaceInfoRepository.deleteByNamespace(nameSpace);
    }

    protected void deleteNameSpaceZooKeeperClusterMappingInDb(String nameSpace) {
        namespaceZkClusterMappingRepository.deleteByNamespace(nameSpace);
    }

    protected void deleteJobsInDB(String nameSpace) {
        currentJobConfigRepository.deleteByNamespace(nameSpace);
    }

    protected void deleteNameSpaceInZooKeeper(NameSpaceConfiguration namespaceConfiguration, String namespace) {
        CuratorFramework curatorFramework = null;
        try {
            curatorFramework = curatorRepository.connect(namespaceConfiguration.getZooKeeperAddressList(), null, namespaceConfiguration.getDigest()
            );
            curatorFramework.delete().deletingChildrenIfNeeded().forPath("/" + namespace);
            log.info("delete namespace in zk success - namespace {}", namespace);
        } catch (Exception e) {
            log.warn("fail to delete namespace:{}", namespace, e);
        } finally {
            if (curatorFramework != null) {
                curatorFramework.close();
            }
        }
    }

}